package net.lainiao;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.recipes.queue.DistributedIdQueue;
import org.apache.curator.framework.recipes.queue.QueueBuilder;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;
import org.apache.curator.utils.CloseableUtils;

/**
 * Created by Administrator on 2016/11/8.
 */
public class DistributedQueueExample2 {
    private static final String PATH="/DistributedQueueExample2/queue";

    public static void main(String[] args) throws Exception {
        TestingServer server=new TestingServer();
        CuratorFramework client=null;
        DistributedIdQueue<String> queue=null;
        try {
            client= CuratorFrameworkFactory.newClient(server.getConnectString(),new ExponentialBackoffRetry(1000,3));
            client.getCuratorListenable().addListener(new CuratorListener() {
                @Override
                public void eventReceived(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                    System.out.println("CuratorEvent:"+ curatorEvent.getType().name());
                }
            });
            client.start();
            QueueConsumer<String> consumer=createConsumer();
            QueueBuilder<String> builder=QueueBuilder.builder(client,consumer,createQueueSerializer(),PATH);
            queue=builder.buildIdQueue();
            queue.start();
            for (int i = 0; i <10 ; i++) {
                String id="Id"+i;
                queue.put("消息"+i,id);
                Thread.sleep((long) (50*Math.random()));
                queue.remove(id);
            }
            Thread.sleep(20000);
        }
        finally {
            CloseableUtils.closeQuietly(queue);
            CloseableUtils.closeQuietly(client);
            CloseableUtils.closeQuietly(server);
        }

    }

    private static QueueSerializer<String> createQueueSerializer() {
        return new QueueSerializer<String>() {
            @Override
            public byte[] serialize(String mess) {
                return mess.getBytes();
            }

            @Override
            public String deserialize(byte[] bytes) {
                return new String(bytes);
            }
        };
    }

    private static QueueConsumer<String> createConsumer() {
        return new QueueConsumer<String>() {
            @Override
            public void consumeMessage(String mess) throws Exception {
                System.out.println("消费这消费了一个消息:"+mess);
            }

            @Override
            public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                System.out.println("连接新状态:"+connectionState.name());
            }
        };
    }
}
